package com.openness.spider.api;

import java.io.Closeable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;

import com.openness.spider.Html;

/**
 * @author Rain
 * @version 2013-6-23 上午11:25:27
 */

public class SpiderDataReader implements Closeable {

	public static final String HBASE_MASTER = "hbase.master";

	public static final String HBASE_ZOOKEEPER_QUORUM = "hbase.zookeeper.quorum";

	public static final String HBASE_ZOOKEEPER_PROPERTY_CLIENTPORT = "hbase.zookeeper.property.clientPort";

	public static final String HBASE_TABLEPOOL_SIZE = "hbase.tablepool.size";

	private HTable table;

	private Scan scan = new Scan();

	private ResultScanner rs;

	public SpiderDataReader(Configuration conf, String tableName,
			byte[] startRow, byte[] stopRow, Filter filter) throws IOException {
		this.table = new HTable(conf, tableName);

		if (startRow != null) {
			scan.setStartRow(startRow);
		}

		if (stopRow != null) {
			scan.setStopRow(stopRow);
		}

		if (filter != null) {
			scan.setFilter(filter);
		}

		rs = table.getScanner(scan);
	}

	public Html next() throws IOException {
		Result result = rs.next();
		if (result == null || result.size() == 0) {
			return null;
		}

		Html html = new Html();

		html.get(result);

		return html;
	}

	public List<Html> next(int max) throws IOException {
		Result[] results = rs.next(max);

		List<Html> htmls = new ArrayList<Html>();

		for (Result result : results) {
			if (result == null || result.size() == 0) {
				continue;
			}

			Html html = new Html();

			html.get(result);

			htmls.add(html);
		}

		return htmls.size() > 0 ? htmls : null;
	}

	@Override
	public void close() throws IOException {
		if (rs != null) {
			rs.close();
		}

		if (table != null) {
			table.close();
		}
	}

	public static void main(String[] args) throws IOException {
		Configuration conf = new Configuration();

		conf.set(SpiderDataReader.HBASE_MASTER, "192.168.1.222:60010");
		conf.set(SpiderDataReader.HBASE_ZOOKEEPER_QUORUM, "192.168.1.222");
		conf.set(SpiderDataReader.HBASE_ZOOKEEPER_PROPERTY_CLIENTPORT, "2181");
		conf.set(SpiderDataReader.HBASE_TABLEPOOL_SIZE, "10");

		SpiderDataReader reader = new SpiderDataReader(conf, "spiderdata",
				null, null, null);

		int count = 0;
		
		List<Html> htmls;

		while ((htmls = reader.next(100)) != null) {
			count += htmls.size();

			System.out.println(count);
		}

		reader.close();
	}

}
